import { Buffer } from 'buffer';
import { IPublishPacket, ISubscribePacket, Packet } from 'mqtt-packet/types/index';
import { Subject } from 'rxjs';
import { generate, parser } from './mqtt.packet.js';
import { excludeKeys, getKey, pool, TIMEOUT } from './pool';

const Parser = parser();

const subject = new Subject();

const MessageID = {
  value: 0,
  next() {
    if (MessageID.value < 65535) {
      MessageID.value++;
    } else {
      MessageID.value = 0;
    }
    return MessageID.value;
  },
};

let pingIntervalID = 0;

let socketTask = null;

function connect() {
  const url = 'wss://mqtt.anlan.tech:8084/mqtt';
  const protocols = ['mqtt'];
  return wx.connectSocket({ url, protocols });
}

function toConnect() {
  if (socketTask !== null) {
    try {
      socketTask.close({
        success() {
          socketTask = connect();
          setEvent();
        },
      });
    } catch (err) {
      console.error(err);
    }
  } else {
    socketTask = connect();
    setEvent();
  }
}

function setEvent() {
  socketTask.onOpen(() => {
    const packet: Packet = {
      cmd: 'connect',
      protocolId: 'MQTT', // Or 'MQIsdp' in MQTT 3.1 and 5.0
      protocolVersion: 4, // Or 3 in MQTT 3.1, or 5 in MQTT 5.0
      clean: true,
      clientId: 'mqttjs_' + Math.random().toString(16).slice(2, 10),
      keepalive: 60, // Seconds which can be any positive number, with 0 as the default setting
      username: 'ZLrSE2Az4wuCmjpI',
      // @ts-ignore
      password: 'BZUS801mDRw4iqI7',
    };
    const data = generate(packet).buffer;
    socketTask.send({ data });
  });

  socketTask.onError(() => {
    clearInterval(pingIntervalID);
    setTimeout(() => {
      socketTask = connect();
      setEvent();
    }, 1000);
  });

  socketTask.onMessage(({ data }) => {
    Parser.parse(<Buffer>data);
  });

  socketTask.onClose(() => {
    clearInterval(pingIntervalID);
    socketTask = connect();
    setEvent();
  });
}

function toClose() {
  socketTask.close({
    success() {
      socketTask = null;
    },
  });
}

function pingreq() {
  const packet: Packet = { cmd: 'pingreq' };
  const data = generate(packet).buffer;
  socketTask.send({ data });
}

function subscribe(topic) {
  const messageId = MessageID.next();

  const key = getKey(topic);
  if (pool[key]) {
    pool[messageId] = pool[key];
  }

  setTimeout(() => {
    if (pool[messageId]) {
      delete pool[messageId];
    }
  }, TIMEOUT);

  const packet: ISubscribePacket = {
    cmd: 'subscribe',
    messageId,
    subscriptions: [{ topic, qos: 0 }],
  };
  const data = generate(packet).buffer;
  socketTask.send({ data });
}

function unsubscribe(topic) {
  const packet: Packet = {
    cmd: 'unsubscribe',
    messageId: MessageID.next(),
    unsubscriptions: [topic],
  };
  const data = generate(packet).buffer;
  socketTask.send({ data });
}

function publish(topic, data) {
  const payload = Buffer.from(JSON.stringify(data));
  const packet: Packet = {
    cmd: 'publish',
    messageId: MessageID.next(),
    qos: 0,
    dup: false,
    topic,
    payload,
    retain: false,
  };
  socketTask.send({
    data: generate(packet).buffer,
    success() {
      const key = getKey(topic);
      if (pool[key]) {
        toNextState(key);
      }
    },
  });
}

function toNextState(key: string) {
  pool[key].machineService.send({ type: 'NEXT' });
}

function onMsg(packet: IPublishPacket) {
  const key = getKey(packet.topic);
  const msg = packet.payload.toString();
  if (excludeKeys.includes(key)) {
    subject.next(JSON.parse(msg));
  } else if (pool[key]) {
    pool[key].onMsg(JSON.parse(msg));
    toNextState(key);
  }
}

Parser.on('packet', (packet) => {
  switch (packet.cmd) {
    case 'connack':
      // @ts-ignore
      pingIntervalID = setInterval(pingreq, 30 * 1000);
      break;
    case 'suback':
      const messageId = packet.messageId.toString();
      if (pool[messageId]) {
        toNextState(messageId);
        delete pool[messageId];
      }
      break;
    case 'unsuback':
      break;
    case 'publish':
      onMsg(packet);
      break;
    case 'puback':
      break;

    default:
      break;
  }
});

toConnect();

export { publish, subscribe, unsubscribe, socketTask, subject };
